package com.lqf.ml.recommand

import java.time.LocalDate

import com.lqf.client.SparkClient
import org.apache.spark.sql.functions.col
import org.apache.spark.sql.{DataFrame, Dataset, Row, SparkSession}

class AdDataServer(ss: SparkSession, hdfsPath: String) {
  val clickCounters = List("ad_adclick", "cp_adclick")
  val viewCounter = Array("ad_adview", "cp_adview")
  val adIDName = "udid"
  val counterName = "counterid"
  val appIDName = "appid"
  val EXT$ip = "ext.ip"
  val hourName = "hour"
  val EXT$net = "ext.net"
  val EXT$system = "ext.system"
  val EXT$adType = "ext.adtype"
  val EXT$display = "ext.display"
  val ADX$ADID = "ext.video_url_md5"
  var noNullCol = Array("deviceID", appIDName, "adID")

  def getBaseDataSet(startDate: LocalDate, endDate: LocalDate, counters: List[String]): DataFrame = {
    val sparkClient = new SparkClient(ss)
    sparkClient.getBaseDataset(hdfsPath, startDate, endDate).filter(col(counterName).isin(counters: _*))
  }

  def getImpressDataForALS(baseDataSet: Dataset[Row]): DataFrame = {
    baseDataSet.filter(col(counterName).equalTo("ad_adview")
      .and(col(EXT$display).equalTo("1").or(col(EXT$display).equalTo(true)))
      .or(col(counterName).equalTo("cp_adview")))
      .filter(col("ext.platform").notEqual("110"))//过滤今日头条广告
      .select("udid", "ext.video_url_md5")
      .withColumnRenamed("udid", "deviceID").withColumnRenamed("video_url_md5", "clickedAd")
      .na.drop(Array("clickedAd"))
  }

  def getClickedDataForALS(baseDataSet: Dataset[Row]): DataFrame = {
    baseDataSet.filter(col(counterName).isin(clickCounters: _*))
      .filter(col("ext.platform").notEqual("110"))//过滤今日头条广告
      .select("udid", "ext.video_url_md5")
      .withColumnRenamed("udid", "deviceID").withColumnRenamed("video_url_md5", "clickedAd")
      .na.drop(Array("clickedAd"))
  }

  def getClickedData(baseDataSet: Dataset[Row]): DataFrame = {
    baseDataSet.filter(col(counterName).isin(clickCounters: _*))
      .select(adIDName, appIDName, ADX$ADID,
        //EXT$adType,现在基本都是2-原生广告，意义不大，未来可以放开增加
        hourName, EXT$net, EXT$system, EXT$ip)
      .withColumnRenamed(adIDName, "deviceID").withColumnRenamed("video_url_md5", "adID").
      na.drop(noNullCol)
  }

  def getImpressionData(baseDataSet: Dataset[Row]): DataFrame = {
    baseDataSet.filter(col(counterName).equalTo("ad_adview")
      .and(col(EXT$display).equalTo("1").or(col(EXT$display).equalTo(true)))
      .or(col(counterName).equalTo("cp_adview")))
      .select(adIDName, appIDName, ADX$ADID,
        //EXT$adType,现在基本都是2-原生广告，意义不大，未来可以放开增加
        hourName, EXT$net, EXT$system, EXT$ip)
      .withColumnRenamed(adIDName, "deviceID").withColumnRenamed("video_url_md5", "adID").
      na.drop(noNullCol)
  }

  def getImpressionData(startDate: LocalDate, endDate: LocalDate): DataFrame = {
    val sparkClient = new SparkClient(ss)
    val data = sparkClient.getBaseDataset(hdfsPath, startDate, endDate)
    data.filter(col(counterName).equalTo("ad_adview")
      .and(col(EXT$display).equalTo("1").or(col(EXT$display).equalTo(true)))
      .or(col(counterName).equalTo("cp_adview")))
      .select(adIDName, appIDName, ADX$ADID,
        //EXT$adType,现在基本都是2-原生广告，意义不大，未来可以放开增加
        hourName, EXT$net, EXT$system, EXT$ip)
      .withColumnRenamed(adIDName, "deviceID").withColumnRenamed("video_url_md5", "adID").
      na.drop(noNullCol)
  }
}
